#pragma once
#include <condition_variable>
#include <string>
#include <functional>
#include <google/protobuf/map_entry_lite.h>
#include <memory>
#include <mutex>
#include <unordered_map>
#include <utility>
#include "muduo/protoc/codec.h"
#include "../mqcommon/Helper.hpp"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "mq_consumer.hpp"
#include "../mqcommon/mq_proto.pb.h"
namespace mq_zxws
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicConsumerResponcePtr = std::shared_ptr<basicConsumerResponce>;
    using basicCommonResponcePtr = std::shared_ptr<basicCommonResponce>;
    class Channal{
    public:
        using ptr = std::shared_ptr<Channal>;
        Channal(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
            :_cid(UUID::uuid()), _conn(conn), _codec(codec)
        {}
        ~Channal(){
            basicCancel();
        }
        std::string cid(){
            return _cid;
        }
        bool openChannal(){
            openChannalRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            _codec->send(_conn, req);
            basicCommonResponcePtr resp = waitResponce(req.rid());
            return resp->ok();
        }
        void closeChannal(){
            closeChannalRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            _codec->send(_conn, req);
            waitResponce(req.rid());
        }
        // 交换机的新增和删除
        bool declareExchange(
            const std::string& ename, 
            const ExchangeType& type,
            bool durable,
            bool auto_delete,
            google::protobuf::Map<std::string, std::string>& args){
            // 设置请求
            declareExchangeRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            basicCommonResponcePtr basic_resp = waitResponce(req.rid());
            return basic_resp->ok();
        }
        void deleteExchange(const std::string& ename){
            deleteExchangeRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            waitResponce(req.rid());
        }

        // 队列的新增和删除
        bool declareQueue(
            const std::string& qname,
            bool exclusive,
            bool durable,
            bool auto_delete,
            google::protobuf::Map<std::string, std::string>& args){
            // 设置请求
            declareQueueRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_exclusive(exclusive);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            basicCommonResponcePtr basic_resp = waitResponce(req.rid());
            return basic_resp->ok();
        }
        void deleteQueue(const std::string& qname){
            deleteQueueRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_queue_name(qname);
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            waitResponce(req.rid());
        }
        // 绑定和解除绑定
        bool queueBind(const std::string& ename, const std::string& qname, const std::string& key){
            // 设置请求
            queueBindRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(key);
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            basicCommonResponcePtr basic_resp = waitResponce(req.rid());
            return basic_resp->ok();
        }
        
        void queueUnBind(const std::string& ename, const std::string& qname){
            queueUnBindRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_queue_name(qname);
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            waitResponce(req.rid());
        }
        // 消息发布
        void basicPublish(const std::string& ename, const BasicProperties* bp, const std::string& body){
            basicPublishRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            if(bp != nullptr){
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_deliver_mode(bp->deliver_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            req.set_body(body);
            DLOG("向 %s 交换机发布 %s", ename.c_str(), body.c_str());
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            waitResponce(req.rid());
        }
        // 消息确认
        void basicAck(const std::string& msg_id){
            if(_consumer.get() == nullptr){
                DLOG("找不到当前的信道的消费者");
                return;
            }
            basicAckRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_message_id(msg_id);
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            waitResponce(req.rid());
        }
        // 订阅消息
        bool basicConsumer(const std::string& consumer_tag, const std::string& qname, bool ack, const ConsumerCallBack& cb){
            if(_consumer.get() != nullptr){
                DLOG("当前信道已经订阅了其他的队列消息");
                return false;
            }
            basicConsumerRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_consumer_tag(consumer_tag);
            req.set_queue_name(qname);
            req.set_auto_ack(ack);
            _codec->send(_conn, req);
            basicCommonResponcePtr resp = waitResponce(req.rid());
            if(!resp->ok()){
                DLOG("订阅失败");
                return false;
            }
            _consumer = std::make_shared<Consumer>(consumer_tag, qname, ack, cb);
            return true;
        }
        // 取消订阅
        void basicCancel(){
            if(_consumer.get() == nullptr){
                return;
            }
            basicCancelRequest req;
            req.set_rid(UUID::uuid());
            req.set_cid(_cid);
            req.set_consumer_tag(_consumer->tag);
            req.set_queue_name(_consumer->qname);
            // 发送请求
            _codec->send(_conn, req);
            // 接收到应答
            waitResponce(req.rid());
            _consumer.reset(); // 置空
        }
        void consumer(const basicConsumerResponcePtr& resp){
            if(_consumer.get() == nullptr){
                DLOG("没有订阅者信息");
                return;
            }
            if(_consumer->tag != resp->consumer_tag()){
                DLOG("收到的推送消息中的消费者标识，与当前消息消费者标识不一致");
                return;
            }
            _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }
        
        void putBasicResponce(const basicCommonResponcePtr& resp){
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(), resp));
            DLOG("唤醒所有等待应答的线程");
            _cv.notify_all();
        }
    private:
        basicCommonResponcePtr waitResponce(const std::string& rid){
            std::unique_lock<std::mutex> lock(_mutex);
            DLOG("等待应答...");
            _cv.wait(lock, [&rid, this](){
                return _basic_resp.find(rid) != _basic_resp.end();
            });
            DLOG("得到应答 rid: %s", rid.c_str());
            basicCommonResponcePtr resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return resp;
        }
    private:
        std::string _cid;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        Consumer::ptr _consumer;
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string, basicCommonResponcePtr> _basic_resp; // [rid, ptr]
    };
    class ChannalManager{
    public:
        using ptr = std::shared_ptr<ChannalManager>;
        ChannalManager(){}
        Channal::ptr create(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec){
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = std::make_shared<Channal>(conn, codec);
            _channals.insert(std::make_pair(it->cid(), it));
            return it;
        }
        void remove(const std::string & cid){
            std::unique_lock<std::mutex> lock(_mutex);
            _channals.erase(cid);
        }
        Channal::ptr getChannal(const std::string & cid){
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channals.find(cid);
            if(it == _channals.end()){
                return Channal::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channal::ptr> _channals;
    };
}